package spark.MLlib;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.mllib.classification.LogisticRegressionModel;
import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS;
import org.apache.spark.mllib.evaluation.MulticlassMetrics;
import org.apache.spark.mllib.regression.LabeledPoint;
import org.apache.spark.mllib.util.MLUtils;
import scala.Tuple2;

/**
 * 作者: LDL
 * 功能说明:
 * 创建日期: 2015/6/30 18:30
 */
public class MultinomialLogisticRegressionExample {

    public static void main(String[] args) {
        System.setProperty("hadoop.home.dir", "F:\\tools\\hadoop-common-2.2.0-bin-master");
        SparkConf conf = new SparkConf().setMaster("local").setAppName("JavaDataTypes");
        JavaSparkContext jsc = new JavaSparkContext(conf);

        //JavaRDD<LabeledPoint> training = MLUtils.loadLibSVMFile(jsc.sc(), "F:\\guideweibo\\weibotrans\\libsvm.txt").toJavaRDD();
        JavaRDD<LabeledPoint> data = MLUtils.loadLibSVMFile(jsc.sc(), "F:\\guideweibo\\weibotest\\libsvm.txt").toJavaRDD();

        //jsc.setLogLevel("OFF");
        // Split initial RDD into two... [60% training data, 40% testing data].
        JavaRDD<LabeledPoint>[] splits = data.randomSplit(new double[]{0.6, 0.4}, 11L);
        JavaRDD<LabeledPoint> training = splits[0].cache();
        JavaRDD<LabeledPoint> test = splits[1];

        /*JavaRDD<LabeledPoint> training = data.sample(false, 0.6, 11L);
        training.cache();
        JavaRDD<LabeledPoint> test = data.subtract(training);*/

        // Run training algorithm to build the model.
        final LogisticRegressionModel model = new LogisticRegressionWithLBFGS()
                .setNumClasses(2)
                .run(training.rdd());

        // Compute raw scores on the test set.
        JavaRDD<Tuple2<Object, Object>> predictionAndLabels = test.map(
                p -> {
                    Double prediction = model.predict(p.features());
                    return new Tuple2<Object, Object>(prediction, p.label());
                }
        );

        // Get evaluation metrics.
        MulticlassMetrics metrics = new MulticlassMetrics(predictionAndLabels.rdd());
        double precision = metrics.precision();
        System.out.println("Precision = " + precision);

        // Save and load model
    }
}
